package wsz.tx;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;

/**
 * 生产者发送事务消息
 * @author wsz
 * @date 2022/2/28 14:23
 **/
public class TxProducer {
    public static void main(String[] args) throws Exception {
        // 事务监听
        TransactionListener transactionListener = new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("执行本地事务");
                try {
                    Thread.sleep(1000);
                } catch (Exception exception) {
                    exception.printStackTrace();
                }

                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                // 如果没有收到生产者发送的Half Message的响应，broker发送请求到生产者回查生产者本地事务的状态
                // 该方法用于获取本地事务执行的状态。
                System.out.println("检查本地事务的状态");

                return LocalTransactionState.ROLLBACK_MESSAGE;
//                return LocalTransactionState.COMMIT_MESSAGE;
            }
        };

        // 绑定事务监听
        TransactionMQProducer mqProducer = new TransactionMQProducer("tx_producer_09");
        mqProducer.setTransactionListener(transactionListener);
        mqProducer.setNamesrvAddr("192.168.174.138:9876");
        mqProducer.start();

        Message message = new Message("tp_demo_09", "tx message".getBytes(StandardCharsets.UTF_8));
        // 参数
        mqProducer.sendMessageInTransaction(message, "");
    }
}
